Pointblank是一款支源多種資料格式的資料驗證套件（註1），由Richard Iannone所開發，並由Posit大力支持。

今天我將分享Pointblank的基礎用法。

本日大綱如下：

0. 安裝Pointblank
1. 核心用法
2. YAML用法
3. Inspection
4. 小結

0. 安裝Pointblank

建議使用uv安裝：

uv add pointblank

習慣上會使用下列語法引入Pointblank：

import pointblank as pb

1. 核心用法

Pointblank核心用法為：

❗ # sudo code
(
    pb.Validate(data=)
    .validation_steps()
    .interrogate()
)

其中：

• pb.Validate() class是Pointblank的起手式，data=為必須給定的資料來源，支援多種資料格式，可以直接傳入pl.DataFrame。
• .validation_steps()為一至多個驗證步驟，函數名大多為.col_*()開頭，例如.col_vals_le()可以用來比較各行是否小於或等於某固定值或某列之值。
• 所有的驗證步驟都是lazy模式，呼叫.interrogate()才正式啟動所有驗證步驟。

以下我們舉兩個例子來實際演示。首先使用pb.load_dataset()讀入Pointblank內建的「"small_table"」資料集，並指定返回型別為pl.DataFrame：

import pointblank as pb
import polars as pl

small_table = pb.load_dataset(dataset="small_table", tbl_type="polars")
small_table.shape

(13, 8)

small_table.columns

['date_time', 'date', 'a', 'b', 'c', 'd', 'e', 'f']

範例一

使用Polars，先計算出「"a"」列中之最大值為8：

small_table.select(pl.col("a").max())

shape: (1, 1)
┌─────┐
│ a   │
│ --- │
│ i64 │
╞═════╡
│ 8   │
└─────┘

如果想要使用.col_vals_le()來驗證「"a"」列中，是否每一行之值皆大於或小於8，可以這麼寫：

pb.Validate(small_table).col_vals_le("a", 8).interrogate()

所得到的結果會是一張使用Great Tables製做而成的表格。除了基本的資訊外，比較值得一提的是：

• 「"UNITS"」列代表所進行的驗證次數，因為共有13行，故顯示為13。
• 「"PASS"」列上方的次數代表通過驗證的次數，下方則是通過驗證的百分比。
• 「"FAIL"」列上方的次數代表沒有通過驗證的次數，下方則是沒有通過驗證的百分比。
• 「"W"」、「"E"」及「"C"」三列則分別代表WARNING、EROOR及CRTITICAL三種警戒層級。此處因為我們沒有給定各警戒層級的門檻值，所以皆顯示為-。
• 「"EXT"」列會將驗證失敗的行數，收集為一個CSV檔，供使用者點擊下載。此處因為所有行數皆驗證成功，所以顯示為-。

範例二

範例二延續範例一。

我們先使用Polars，計算出「"f"」列內的三個元素分別為「"mid"」、「"low"」及「"high"」：

small_table.select(pl.col("f").unique())

shape: (3, 1)
┌──────┐
│ f    │
│ ---  │
│ str  │
╞══════╡
│ mid  │
│ low  │
│ high │
└──────┘

接著執行以下程式碼：

def send_alert():
    summary = pb.get_validation_summary()
    if summary["highest_severity"] == "critical":
        print(
            f"ALERT: Critical validation failures found in {summary['tbl_name']}"
        )


(
    pb.Validate(
        data=small_table,
        tbl_name="Small Table",
        label="Polars熊霸天下[Day28]",
        thresholds=pb.Thresholds(warning=0.1, error=0.2, critical=0.3),
        actions=pb.Actions(error="ERROR DETECTED: {type}"),
        final_actions=pb.FinalActions(send_alert),
        brief=True,
        lang="zh-Hant",
    )
    .col_vals_le("a", 6)
    .col_vals_in_set(
        "f",
        ["mid", "high"],
        actions=pb.Actions(warning="WARNING DETECTED: {type}"),
    )
    .interrogate()
)

WARNING DETECTED: col_vals_in_set
ALERT: Critical validation failures found in Small Table

這次的行數雖然較前例為多，但基本架構還是由pb.Validate()、.validation_steps()及.interrogate()所組成，以下分別說明。

pb.Validate()

pb.Validate()可以視為整個驗證流程的全域設定。

在範例二中，我們進行了以下設定：

• 指定data=為small_table dataframe。
• 指定tbl_name=為「"Small Table"」（顯示於標頭）。
• 指定label=為「"Polars熊霸天下[Day28]"」（顯示於標頭）。
• 使用pb.Thresholds設定三個層級的警戒值，並指定給thresholds=。這邊需注意，使用者可以僅指定需要的層級，不需要三個都指定。
• 使用pb.Actions設定全域內之「"ERROR"」層級被觸發時之動作，並指定給actions=。其中的{type}是Pointblank所提供的字串模板之一，詳細列表可以參考API文件。另外，每一驗證步驟完成後即會立即觸發動作，而不是等所有步驟完成後才一次觸發。
• 使用pb.FinalActions設定此驗證流程完成時，所需觸發的動作（呼叫send_alert()），並指定給FinalActions=。
• 指定brief=為True，自動產生各驗證步驟的概述。
• 指定lang=為「"zh-Hant"」，設定表格語言為正體中文。

.validation_steps()

設立了兩個validation steps：

• 使用.col_vals_le()來驗證「"a"」列中，是否每一行之值皆大於或小於6。
• 使用.col_vals_in_set()來驗證「"f"」列中元素是否都在「"mid"」及「"high"」中。請注意，此處使用pb.Actions設定此驗證步驟之「"WARNING"」層級被觸發時之動作，並指定給actions=。

.interrogate()

呼叫.interrogate()後，才會開始所有驗證步驟。

討論

範例二表格與範例一表格相比豐富了許多。

比較特別的是「"W"」、「"E"」及「"C"」三列變為圓心，其中空心及實心，分別代表沒有達到及已經達到警戒值。也就是說：

• 在第一個步驟中，.col_vals_le()已經達到「"WARNING"」的警戒值。但需注意，由於全域及此步驟內都未設定「"WARNING"」的動作，所以沒有動作會被觸發。
• 在第二個步驟中，三種層級的警戒值都已達到。由於此步驟有設定pb.Actions(warning="WARNING DETECTED: {type}")動作，所以將會覆蓋全域預設的pb.Actions(error="ERROR DETECTED: {type}")。

此外，點擊「"EXT"」列的CSV藍色按紐，可以下載該步驟失敗的行數。以下為第一個步驟的CSV檔案內容：

_row_num_,date_time,date,a,b,c,d,e,f
5,2016-01-09T12:36:00.000000,2016-01-09,8,3-ldm-038,7,283.94,true,low
7,2016-01-15T18:46:00.000000,2016-01-15,7,1-knw-093,3,843.34,true,high

可以看出此兩行的「"a"」列值為8及7，皆沒有小於或等於6，所以沒有通過驗證。

2. YAML用法

除了核心寫法外，Pointblank還提供了使用YAML來編寫驗證流程的寫法，教學文件中提供了各種細項設定說明。這將使得版控更為容易，讓團隊或專案能有統一的驗證流程。

舉例來說，我們可以將範例一改用以下的YAML格式表達，存為small_table.yaml：

tbl: small_table
steps:
  - col_vals_le:
      columns: a
      value: 8

接著使用pb.yaml_interrogate()來執行：

yaml_file = "small_table.yaml"
pb.yaml_interrogate(yaml_file)

可以得到與核心用法相同的表格。

最後，使用pb.yaml_to_python()可以印出YAML檔相對應程式碼，例如：

pb.yaml_to_python(yaml_file)

import pointblank as pb

(
    pb.Validate(data=pb.load_dataset("small_table", tbl_type="polars"))
    .col_vals_le(columns="a", value=8)
    .interrogate()
)

這將使得debug更為容易。

3. Inspection

Inspection也是Pointblank的強項，以下介紹三個我最常用的函數，分別是pb.preview()、pb.col_summary_tbl()及pb.missing_vals_tbl()。

特別提醒，雖然很多套件都會提供類似的功能，但這些函數支援多種資料格式，方便使用一致的寫法處理各種資料。

pb.preview()

pb.preview()預設會顯示前五行及後五行，例如：

pb.preview(small_table)

可以使用columns_subset=來選定部份列及指定適合的n_head=或n_tail=來顯示行數，例如：

pb.preview(
    small_table,
    columns_subset=["date_time", "date", "a", "b", "c"],
    n_head=2,
    n_tail=2,
)

pb.col_summary_tbl()

pb.col_summary_tbl()可以快速列出各列的統計資訊。

pb.col_summary_tbl(small_table)

pb.missing_vals_tbl()

pb.missing_vals_tbl()可以快速列出缺失值所在的大概位置。這對於含有時間序列的資料相當有用，因為將可以用時間思考缺失的可能原因。

pb.missing_vals_tbl(small_table)

4. 小結

本日只介紹了Pointblank的皮毛，其它進階用法還有CLI及MCP等。

值得一提的是，作者花費了很多心力編寫說明文件，使其簡潔易懂。有興趣的朋友可以參考他們的部落格文章，深入了解其文件設計理念。

備註

註1：Pointblank底層使用narwhals及ibis來處理多種資料格式。除了Pointblank之外，Python還有許多類似的套件，有興趣的朋友可以參考Pointblank的部落格文章。該篇文章比較了包含Pointblank在內，五種支援Polars的資料驗證套件。

Code

本日程式碼傳送門。